package com.wormpex.data.root.service.impl;


import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.wormpex.data.root.dao.FromToEntryDao;
import com.wormpex.data.root.model.bean.FromToEntry;
import com.wormpex.data.root.service.FromToMappingService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;

/**
 * Created by ouryousa on 2017/8/17.
 */

@Service
public class FromToMappingServiceImpl implements FromToMappingService{
    private static Logger logger = LoggerFactory.getLogger(FromToMappingServiceImpl.class);
    private volatile Map<String, List<FromToEntry>> wmqkafkaMapping; //key:wmq的topic_name

    @Autowired
    private FromToEntryDao fromToEntryDao;

    @PostConstruct
    public void init(){
        //更新映射关系
        wmqkafkaMapping = getWmqKafkaMapping();
    }

    /**
     * 定时更新wmq和kafka的映射关系
     */
    @Scheduled(cron = "0 0/15 * * * ?")
    private void refresh() {
        // 更新缓存信息
        try {
            this.wmqkafkaMapping = getWmqKafkaMapping();
        } catch (Exception e) {
            logger.warn("update from_to_mapping failed");
        }
    }

    public Map<String, List<FromToEntry>> getWmqKafkaMapping(){
        Map<String, List<FromToEntry>> fromToEntryMap = Maps.newHashMap();
        List<FromToEntry> fromToEntries = fromToEntryDao.selectWmqKafka();
        if (fromToEntries.size() == 0) {
            throw new RuntimeException("Can't get fromToMapping from db");
        }
        for (FromToEntry fromToEntry : fromToEntries) {
            List<FromToEntry> fromToEntryList = null;
            if(fromToEntryMap.keySet().contains(fromToEntry.getFromTopicName())) {
                fromToEntryList = fromToEntryMap.get(fromToEntry.getFromTopicName());
            }else{
                fromToEntryList = Lists.newArrayList();
                fromToEntryMap.put(fromToEntry.getFromTopicName(), fromToEntryList);
            }
            fromToEntryList.add(fromToEntry);
        }
        return fromToEntryMap;
    }

    @Override
    public List<FromToEntry> getWmqKafkaEntries(String FromTopicName) {
        return wmqkafkaMapping.get(FromTopicName);
    }
}
